/**
 * @(#)KafkaMqManager.java 2017年3月28日
 * Copyright(C) 2017年3月28日 ZHEXIN IT CO.,LTD. All rights reserved.
 */
package org.jsbd.boss.mq;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.jsbd.boss.domian.DMClientInfo;
import org.jsbd.boss.domian.SimulatorInfoVo;
import org.jsbd.boss.domian.channel.SimulatorActionLog;
import org.jsbd.boss.domian.channel.SmsResultVo;
import org.jsbd.boss.domian.statistics.BaseDTO;
import org.jsbd.boss.domian.statistics.LostImsiStatistics;
import org.jsbd.boss.domian.statistics.SdkGetSmsStatistics;
import org.jsbd.boss.domian.statistics.SdkSyncSmsStatistics;
import org.jsbd.boss.domian.statistics.SimulatorReqStatistics;
import org.jsbd.boss.domian.statistics.SimulatorSyncStatistics;
import org.jsbd.boss.enums.TaskActionEnum;
import org.jsbd.boss.enums.TaskStatusEnum;
import org.jsbd.boss.enums.TaskTypeEnum;
import org.jsbd.boss.task.BaseTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.webnion.log.api.bean.DbTaskBean;

/**
 * kafka消息队列管理类
 * 
 * @author <a href="mailto:zhouchao@zhexinit.com" >周超</a>
 * @version 1.0.0
 */
public class KafkaMqManager extends BaseTask {
	public static final Logger LOGGER = LoggerFactory.getLogger(KafkaMqManager.class);

	private static AtomicInteger count = new AtomicInteger();

	private static BlockingQueue<BaseDTO<Object>> dataQueue = new LinkedBlockingQueue<BaseDTO<Object>>(10000);

	private Producer<String, String> producer;
	private String topic;
	private static String logTopic;
	private static boolean addTaskFlag = false;

	public static void addTask(BaseDTO<Object> object) {
		if (!addTaskFlag) {
			return;
		}
		// 积累了一万就丢弃
		int total = count.getAndIncrement();
		if (total > 10000) {
			count.decrementAndGet();
			LOGGER.error("total [{}] is so much", total);
			return;
		}
		dataQueue.add(object);
	}

	public void start() {
		Properties props = new Properties();
		try {
			props.load(ClassLoader.getSystemResourceAsStream("kafkaProductor.properties"));
			this.topic = props.getProperty("topic", "monitor");
			logTopic=props.getProperty("log.topic", "logsave");
			producer = new KafkaProducer<String, String>(props);
		} catch (Exception e) {
			LOGGER.error("", e);
		}
	}

	@Override
	public void run() {
		addTaskFlag = true;
		start();
		long i = 0;
		while (true) {
			BaseDTO<Object> data;
			try {
				data=dataQueue.poll(100,TimeUnit.MILLISECONDS);
				if(data==null) {
					producer.flush();
					data = dataQueue.take();
				}
				
				count.getAndDecrement();
				i++;
				
				String topic = data.getTopic();
				if (StringUtils.isBlank(topic)) {
					topic = getTopic();
				}
				
				
				String dataStr=null;
				if("log".equals(data.getActionType())) {
					dataStr=(String)data.getData();
				}else {
					dataStr=JSON.toJSONString(data);
				}
				LOGGER.info("totalCount is [{}]", i);
				producer.send(new ProducerRecord<String, String>(topic, dataStr));
//				producer.flush();
			} catch (InterruptedException e1) {
				LOGGER.error("", e1);
			} catch (Exception e) {
				LOGGER.error("", e);
				if (producer != null) {
					producer.close();
				}
				start();
			}
		}
	}

	public String getTopic() {
		return topic;
	}

	public void setTopic(String topic) {
		this.topic = topic;
	}

	/**
	 * 
	 * 将获取短信的情况发送给MQ
	 * 
	 * @author <a href="mailto:zhouchao@zhexinit.com" >周超</a>
	 * @param loginResult
	 * @param clientInfo
	 */
	public static void getSmsToQueue(SmsResultVo loginResult, DMClientInfo clientInfo) {
		if (!addTaskFlag) {
			LOGGER.debug("功能未开放");
			return;
		}
		SdkGetSmsStatistics sms = new SdkGetSmsStatistics();
		sms.setImsi(clientInfo.getImsi());
		sms.setMark(clientInfo.getMark());
		if (loginResult.getLoginType() != null) {
			sms.setLoginType(loginResult.getLoginType().getType());
		}

		sms.setLogType(TaskActionEnum.SEND_TASK.getType());
		sms.setTaskType(TaskTypeEnum.ADD_TASK.getType());
		sms.setSuccess(loginResult.getTaskStatus() == TaskStatusEnum.SUCCESS);
		sms.setProvince(clientInfo.getProvince());
		BaseDTO<Object> data = new BaseDTO<Object>();
		data.setActionType("1");
		data.setData(sms);
		KafkaMqManager.addTask(data);
	}

	/**
	 * 
	 * 将获取短信的情况发送给MQ
	 * 
	 * @author <a href="mailto:zhouchao@zhexinit.com" >周超</a>
	 * @param clientInfo
	 */
	public static void syncSmsToQueue(DMClientInfo clientInfo) {
		if (!addTaskFlag) {
			LOGGER.debug("功能未开放");
			return;
		}
		SdkSyncSmsStatistics sms = new SdkSyncSmsStatistics();
		sms.setImsi(clientInfo.getImsi());
		sms.setMark(clientInfo.getMark());
		sms.setLoginType(clientInfo.getLoginType());
		sms.setLogType(clientInfo.getActionType());
		sms.setTaskType(clientInfo.getTaskType());
		sms.setProvince(clientInfo.getProvince());

		if ("0".equals(clientInfo.getResultCode())) {
			sms.setSendSuccuss(true);
		} else {
			sms.setSendSuccuss(false);
		}

		if (TaskStatusEnum.SUCCESS.getType() == clientInfo.getStatus()) {
			sms.setInsertTaskSuccess(true);
		} else {
			sms.setInsertTaskSuccess(false);
		}

		BaseDTO<Object> data = new BaseDTO<Object>();
		data.setActionType("2");
		data.setData(sms);
		KafkaMqManager.addTask(data);
	}

	/**
	 * | 将模拟器请求记录到mq
	 * 
	 * @author <a href="mailto:zhouchao@zhexinit.com" >周超</a>
	 * @param simulator
	 */
	public static void simulatorReqToQueue(SimulatorInfoVo simulator) {
		if (!addTaskFlag) {
			LOGGER.debug("功能未开放");
			return;
		}
		SimulatorReqStatistics req = new SimulatorReqStatistics();
		req.setBusyFlag(simulator.isBusy());
		req.setReqTime(new Date());
		req.setUuid(simulator.getUuid());
		req.setVersion(simulator.getSysVer());

		BaseDTO<Object> data = new BaseDTO<Object>();
		data.setActionType("3");
		data.setData(req);
		KafkaMqManager.addTask(data);
	}

	/**
	 * 
	 * 将模拟器任务结果回传到Mq
	 * 
	 * @author <a href="mailto:zhouchao@zhexinit.com" >周超</a>
	 */
	public static void simulatorSyncToQueue(DMClientInfo taskInfo) {
		if (!addTaskFlag) {
			LOGGER.debug("功能未开放");
			return;
		}
		SimulatorSyncStatistics sync = new SimulatorSyncStatistics();
		sync.setUuid(taskInfo.getUuid());
		sync.setSyncTime(new Date());
		BaseDTO<Object> data = new BaseDTO<Object>();
		data.setActionType("4");
		data.setData(sync);
		addTask(data);

		returnTaskTimeToQueue(taskInfo);
	}

	/**
	 * 
	 * 丢失用户数统计
	 * 
	 * @author <a href="mailto:zhouchao@zhexinit.com" >周超</a>
	 */
	public static void lostImsiToQueue(String imsi) {
		if (!addTaskFlag) {
			LOGGER.debug("功能未开放");
			return;
		}

		LostImsiStatistics lost = new LostImsiStatistics();
		lost.setImsi(imsi);
		BaseDTO<Object> data = new BaseDTO<Object>();
		data.setActionType("5");
		data.setData(lost);
		addTask(data);
	}

	/**
	 * 
	 * 模拟器任务日志统计
	 * 
	 * @author <a href="mailto:zhouchao@zhexinit.com" >周超</a>
	 * @param log
	 */
	public static void simulatorLogToQueue(SimulatorActionLog log) {
		if (!addTaskFlag) {
			LOGGER.debug("功能未开放");
			return;
		}

		BaseDTO<Object> data = new BaseDTO<Object>();
		data.setActionType("6");
		data.setData(log);
		addTask(data);
	}

	/**
	 * 
	 * 统计任务耗时-获取任务时间
	 * 
	 * @author <a href="mailto:zhouchao@zhexinit.com" >周超</a>
	 */
	public static void reqTaskTimeToQueue(String task) {
		if (!addTaskFlag) {
			LOGGER.debug("功能未开放");
			return;
		}

		if (task == null) {
			return;
		}

		try {
			JSONObject object = JSON.parseObject(task);
			Map<String, String> dataMap = new HashMap<String, String>();
			dataMap.put("imsi", object.getString("imsi"));
			dataMap.put("mark", object.getString("mark"));
			dataMap.put("time", System.currentTimeMillis() + "");
			Integer taskType = object.getInteger("taskType");
			if (taskType == null) {
				taskType = -1;
			}
			dataMap.put("taskType", taskType + "");

			Integer baseId = object.getInteger("base");
			if (baseId == null) {
				baseId = -9999;
			}
			dataMap.put("baseId", baseId + "");

			BaseDTO<Object> data = new BaseDTO<Object>();
			data.setActionType("1");
			data.setData(dataMap);
			data.setTopic("costTime");
			KafkaMqManager.addTask(data);
		} catch (Exception e) {
			LOGGER.error("reqTaskTimeToQueue error", e);
		}

	}

	/**
	 * 统计任务耗时-任务回传时间
	 * 
	 * @author <a href="mailto:zhouchao@zhexinit.com" >周超</a>
	 */
	public static void returnTaskTimeToQueue(DMClientInfo clientInfo) {
		if (!addTaskFlag) {
			LOGGER.debug("功能未开放");
			return;
		}
		if (clientInfo == null) {
			return;
		}

		Map<String, String> dataMap = new HashMap<String, String>();
		dataMap.put("imsi", clientInfo.getImsi());
		dataMap.put("mark", clientInfo.getMark());
		dataMap.put("time", System.currentTimeMillis() + "");

		BaseDTO<Object> data = new BaseDTO<Object>();
		data.setActionType("2");
		data.setData(dataMap);
		data.setTopic("costTime");
		KafkaMqManager.addTask(data);
	}

	/**
	 * 
	 * 日志保存
	 * @author <a href="mailto:zhouchao@zhexinit.com" >周超</a>
	 */
	public static void saveLogToQueue(String task) {
		if (!addTaskFlag) {
			LOGGER.debug("功能未开放");
			return;
		}

		if (task == null) {
			return;
		}

		try {
			BaseDTO<Object> data = new BaseDTO<Object>();
			data.setActionType("log");
			data.setData(task);
			data.setTopic(logTopic);
			KafkaMqManager.addTask(data);
		} catch (Exception e) {
			LOGGER.error("保存日志 error", e);
		}
	}
	
	/**
	 * 
	 * 用户信息日志保存
	 * @author <a href="mailto:zhouchao@zhexinit.com" >周超</a>
	 */
	public static void saveUserLogToQueue(String task) {
		if (!addTaskFlag) {
			LOGGER.debug("功能未开放");
			return;
		}

		if (task == null) {
			return;
		}

		try {
			BaseDTO<Object> data = new BaseDTO<Object>();
			data.setData(task);
			data.setTopic("userLog");
			KafkaMqManager.addTask(data);
		} catch (Exception e) {
			LOGGER.error("保存日志 error", e);
		}
	}
	
	/**
	 * 
	 * 信息保存
	 * @author <a href="mailto:zhouchao@zhexinit.com" >周超</a>
	 */
	public static void saveDtoToQueue(BaseDTO<Object> data) {
		if (!addTaskFlag) {
			LOGGER.debug("功能未开放");
			return;
		}

		if (data == null) {
			return;
		}

		try {
			data.setActionType("log");
			KafkaMqManager.addTask(data);
		} catch (Exception e) {
			LOGGER.error("saveDtoToQueue error", e);
		}
	}
	
	public static void saveDbTaskToQueue(DbTaskBean dbTask) {
		if (!addTaskFlag) {
			LOGGER.debug("功能未开放");
			return;
		}

		if (dbTask == null) {
			return;
		}

		try {
			BaseDTO<Object> data=new BaseDTO<>();
			data.setActionType("log");
			data.setTopic(logTopic);
			data.setData(JSON.toJSONString(dbTask));
			KafkaMqManager.addTask(data);
		} catch (Exception e) {
			LOGGER.error("saveDtoToQueue error", e);
		}
	}
	
}
